package it.wyx.kafkaclients.consumer;

/**
 * ClassName KafkaConsumerBacklog
 * Description 积压数据处理，判断积压后直接发信分区
 * @author qq
 * @date 2025/4/25 14:06
 * @since JDK 1.8
 */
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerWithBacklogBypass {

    private static final String SOURCE_TOPIC = "source-topic";
    private static final String TARGET_TOPIC = "target-topic";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String GROUP_ID = "bypass-group";
    private static final long BACKLOG_THRESHOLD = 1000; // 积压阈值（消息数量）

    public static void main(String[] args) {
        // 1. 初始化生产者
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(producerProps);

        // 2. 初始化消费者
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 手动提交偏移量
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Collections.singletonList(SOURCE_TOPIC));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    continue;
                }

                // 获取当前分区的最新偏移量
                TopicPartition partition = new TopicPartition(SOURCE_TOPIC, records.partitions().iterator().next().partition());
                long endOffset = consumer.endOffsets(Collections.singleton(partition)).get(partition);
                long currentOffset = records.records(partition).get(records.records(partition).size() - 1).offset();

                // 判断积压是否超过阈值
                if (endOffset - currentOffset > BACKLOG_THRESHOLD) {
                    System.out.println("检测到积压，跳过当前批次，直接消费新消息");
                    // 手动将偏移量提交到最新位置（跳过积压）
                    consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(endOffset + 1)));
                    continue;
                }

                // 处理当前批次消息（无积压时）
                for (ConsumerRecord<String, String> record : records) {
                    // 模拟处理逻辑（此处直接转发到新Topic）
                    ProducerRecord<String, String> newRecord = new ProducerRecord<>(TARGET_TOPIC, record.key(), record.value());
                    producer.send(newRecord);
                }

                // 手动提交偏移量
                consumer.commitSync();
            }
        } finally {
            producer.close();
            consumer.close();
        }



    }
}